package edu.xhu.behavior.listener;

import cn.hutool.json.JSONArray;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson2.JSON;
import com.xxl.job.core.handler.annotation.XxlJob;
import edu.xhu.behavior.mapper.BiliUserChatDetailMapper;
import edu.xhu.behavior.mapper.BiliUserChatMapper;
import edu.xhu.common.exception.CustomException;
import edu.xhu.model.behavior.enums.MessageStatus;
import edu.xhu.model.behavior.pojos.BiliUserChatDetail;
import edu.xhu.model.behavior.vo.MessageRecallVo;
import edu.xhu.model.common.constants.kafka.group.ChatConsumerGroup;
import edu.xhu.model.common.constants.kafka.topic.ChatTopic;
import edu.xhu.model.common.enums.AppHttpCodeEnum;
import edu.xhu.model.common.enums.IsFlagEnum;
import edu.xhu.model.im.enums.IMSendCode;
import edu.xhu.model.im.pojos.IMRecvInfo;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.interceptor.TransactionAspectSupport;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;

/**
 * IM回调消息处理
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class ImRecallListener {

    //同步缓存,存储还未一致性的id:code
    private static final ConcurrentHashMap<String, String> detailIdMaps = new ConcurrentHashMap<>();

    private final BiliUserChatDetailMapper biliUserChatDetailMapper;

    private final BiliUserChatMapper biliUserChatMapper;


    //监听IM回调
    @KafkaListener(topics = ChatTopic.BILL_CHAT_IM_PROCESS, groupId = ChatConsumerGroup.BILI_CHAT_GROUP)
    @Transactional
    public void detail_save(List<String> messages,
                            @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                            @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics) {
        log.info("开始处理im回调");
        long start = System.currentTimeMillis();
        //回调消息处理
        if (messages != null && messages.size() > 0) {
            List<MessageRecallVo> vos = new ArrayList<>();      //在线消息,直接更新送达
            List<MessageRecallVo> offs = new ArrayList<>();     //离线消息(推送前在线！推送后离线),更新离线状态为1,送达状态为2
            List<MessageRecallVo> pullOffs = new ArrayList<>(); //推送的离线消息(推送前就离线),更新离线状态为0,送达状态为2
            messages.forEach(str -> {
                IMRecvInfo result = JSONUtil.toBean(str, IMRecvInfo.class);
                // TODO,注意,result有两种状态,detailId为null时是离线消息,不为null时是具体消息;isRdb表明是否已持久化过
                if (result.getDetailId() != null) {
                    //map中,总共就两种状态,true or false
                    //如果发送成功,加入到发送成功列表
                    if (result.getCode().equals(IMSendCode.SUCCESS.code())) {
                        //需要修改状态为已送达,添加到vos
                        MessageRecallVo vo = new MessageRecallVo();
                        vo.setDetailId(result.getDetailId())
                                .setUserId(result.getSender().getId())
                                .setToId(result.getReceivers().get(0).getId());
                        vos.add(vo);
                    }
                    //如果对方不在线,离线处理(更新消息off状态)
                    else if (result.getCode().equals(IMSendCode.NOT_ONLINE.code())) {
                        //需要修改离线状态,添加到offs
                        MessageRecallVo off = new MessageRecallVo();
                        off.setDetailId(result.getDetailId())
                                .setUserId(result.getSender().getId())
                                .setToId(result.getReceivers().get(0).getId());
                        offs.add(off);
                    }
                    //如果出错了
                    else {
                        //设置为false,定时任务一波清理走
                        detailIdMaps.put(result.getDetailId().toString(), "false");
                    }
                } else {
                    //是离线消息,获取离线data
                    String jsonList = result.getData().toString();
                    JSONArray jsonArray = JSONUtil.parseArray(jsonList);
                    List<BiliUserChatDetail> offLineDetails = jsonArray.stream().parallel().map(object -> JSONUtil.toBean((JSONObject) object, BiliUserChatDetail.class)).collect(Collectors.toList());
                    if (!offLineDetails.isEmpty()) {
                        //如果发送成功,update状态,其他不做处理
                        if (result.getCode().equals(IMSendCode.SUCCESS.code())) {
                            List<MessageRecallVo> pullVos = offLineDetails.stream().parallel().map(detail -> {
                                MessageRecallVo pullOff = new MessageRecallVo();
                                pullOff.setDetailId(detail.getDetailId())
                                        .setUserId(detail.getUserId())
                                        .setToId(detail.getToId());
                                return pullOff;
                            }).collect(Collectors.toList());
                            pullOffs.addAll(pullVos);
                        }
                    }
                }
            });
            //vos有东西,执行update,修改状态为已送达
            if (!vos.isEmpty()) {
                //执行方法
                try {
                    doUpdateForDetailStatus(vos, MessageStatus.SENDED.getCode(), true);
                } catch (CustomException e) {
                    log.error("更新状态失败,不知道为什么欸{}", e.toString());
                    //重试！
                    log.info("开始尝试重试!");
                    try {
                        doUpdateForDetailStatus(vos, MessageStatus.SENDED.getCode(), true);
                    } catch (CustomException again) {
                        log.error("再次更新失败,请检查一下db！{}", again.toString());
                        TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();  //标记为回滚
                    }
                }
            }
            //offs有东西,执行update,修改状态为离线
            if (!offs.isEmpty()) {
                try {
                    doUpdateForDetailStatus(offs, IsFlagEnum.YES.getCode(), true);
                } catch (CustomException e) {
                    log.error("更新离线状态失败,不知道为什么欸{}", e.toString());
                    //重试！
                    log.info("开始尝试重试!");
                    try {
                        doUpdateForDetailStatus(offs, IsFlagEnum.YES.getCode(), true);
                    } catch (CustomException again) {
                        log.error("再次更新失败,请检查一下db！{}", again.toString());
                        TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();  //标记为回滚
                    }
                }
            }
            //pullOffs有东西,执行update,修改状态
            if (!pullOffs.isEmpty()) {
                try {
                    doUpdateForDetailStatus(pullOffs, IsFlagEnum.NO.getCode(), false);
                } catch (CustomException e) {
                    log.error("更新离线状态失败,不知道为什么欸{}", e.toString());
                    //重试！
                    log.info("开始尝试重试!");
                    try {
                        doUpdateForDetailStatus(pullOffs, IsFlagEnum.NO.getCode(), false);
                    } catch (CustomException again) {
                        log.error("再次更新失败,请检查一下db！{}", again.toString());
                        TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();  //标记为回滚
                    }
                }
            }
        }
        long end = System.currentTimeMillis();
        log.info("im回调耗时:{}ms", (end - start));
    }

    //监听rdb回调
    @KafkaListener(topics = ChatTopic.BILL_CHAT_RDB_PROCESS, groupId = ChatConsumerGroup.BILI_CHAT_GROUP)
    @Transactional
    public void detail_rdb(List<String> messages,
                           @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                           @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics) {
        log.info("开始处理rdb回调");
        long start = System.currentTimeMillis();
        //回调rdb消息处理
        if (messages != null && messages.size() > 0) {
            messages.forEach(str -> {
                JSONArray array = JSONUtil.parseArray(str);
                //已经持久化好的消息详情
                List<BiliUserChatDetail> rdbDetails = array.stream().map(bean -> JSONUtil.toBean((JSONObject) bean, BiliUserChatDetail.class)).collect(Collectors.toList());
                List<MessageRecallVo> vos = rdbDetails.stream().parallel()
                        .filter(detail -> detail.getIsOffline().equals(IsFlagEnum.NO.getCode()))    //过滤掉离线消息
                        .map(detail -> {
                            MessageRecallVo messageRecallVo = new MessageRecallVo();
                            BeanUtils.copyProperties(detail, messageRecallVo);
                            return messageRecallVo;
                        }).collect(Collectors.toList());
                if (!vos.isEmpty()) {
                    //执行方法
                    try {
                        doUpdateForDetailStatus(vos, MessageStatus.SENDED.getCode(), true);
                    } catch (CustomException e) {
                        log.error("更新状态失败,不知道为什么欸{}", e.toString());
                        //重试！
                        log.info("开始尝试重试!");
                        try {
                            doUpdateForDetailStatus(vos, MessageStatus.SENDED.getCode(), true);
                        } catch (CustomException again) {
                            log.error("再次更新失败,请检查一下db！{}", again.toString());
                            TransactionAspectSupport.currentTransactionStatus().setRollbackOnly();  //标记为回滚
                        }
                    }
                }
            });
        }
        long end = System.currentTimeMillis();
        log.info("rdb回调耗时:{}ms", (end - start));
    }

    //监听已读回执
    @KafkaListener(topics = ChatTopic.BILL_CHAT_READED_PROCESS, groupId = ChatConsumerGroup.BILI_READ_GROUP)
    @Transactional
    public void readed(List<String> messages,
                       @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                       @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics) {
        log.info("开始处理已读回调");
        long start = System.currentTimeMillis();
        List<Long> senderListForCleanUnRead = new ArrayList<>();
        List<Long> recvListForCleanUnRead = new ArrayList<>();
        List<BiliUserChatDetail> allUnRead = new ArrayList<>();
        //1.回调已读消息处理
        if (messages != null && messages.size() > 0) {

            messages.forEach(str -> {
                IMRecvInfo result = JSON.to(IMRecvInfo.class, str);
                //如果没有报错,更新为已读！
                if (!result.getCode().equals(IMSendCode.UNKONW_ERROR.code())) {
                    //如果是推送给自己,表明需要清除未读
                    if (result.getDetailId() != null) {
                        senderListForCleanUnRead.add(result.getDetailId());
                        recvListForCleanUnRead.add(result.getReceivers().get(0).getId());
                    }
                    //如果不是,表明是传达已读回执,修改消息状态
                    else {
                        List<BiliUserChatDetail> unReadMessage = JSON.parseArray(result.getData().toString()).stream().parallel()
                                .map(json -> JSON.to(BiliUserChatDetail.class, json)).collect(Collectors.toList());
                        if (!unReadMessage.isEmpty() && !allUnRead.addAll(unReadMessage)) {
                            log.info("添加失败~为什么呢?");
                            throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
                        }
                    }
                }
            });
        }
        //2.处理清除未读
        //2.1如果都不为空
        if (!senderListForCleanUnRead.isEmpty() && !recvListForCleanUnRead.isEmpty()) {
            //2.2 size相同
            if (senderListForCleanUnRead.size() == recvListForCleanUnRead.size()) {
                Map<String, List<Long>> ids = new HashMap<>();
                try {
                    //一波处理
                    ids.put("senders", senderListForCleanUnRead);
                    ids.put("recv", recvListForCleanUnRead);
                    if (biliUserChatMapper.updateForCleanUnRead(ids) < 1) {
                        throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
                    }
                } catch (Exception e) {
                    log.info("清除失败~再尝试一次吧~{}", e.toString());
                    try {
                        //一波处理
                        if (biliUserChatMapper.updateForCleanUnRead(ids) < 1) {
                            throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
                        }
                    } catch (CustomException again) {
                        log.info("还是清除失败~{}", again.toString());
                        throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
                    }
                }
            }
            //2.3 size不同(报错咯)
            else {
                log.info("为什么size不同啊?");
                throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
            }
        }
        //3.处理已读状态
        if (!allUnRead.isEmpty()) {
            List<Long> ids = new ArrayList<>();
            try {
                ids = allUnRead.stream().parallel().map(BiliUserChatDetail::getDetailId).collect(Collectors.toList());
                if (biliUserChatDetailMapper.updateByIdBatch(ids, MessageStatus.READED.getCode()) < 1) {
                    throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
                }
            } catch (CustomException e) {
                log.info("更新失败咯{}", e.toString());
                try {
                    if (biliUserChatDetailMapper.updateByIdBatch(ids, MessageStatus.READED.getCode()) < 1) {
                        throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
                    }
                } catch (CustomException again) {
                    log.info("再次更新失败咯{}", e.toString());
                    throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
                }
            }
        }
        long end = System.currentTimeMillis();
        log.info("rdb回调耗时:{}ms", (end - start));
    }

    /**
     * 同步方法:根据code维护不同状态
     * (有没有一种可能,rdb过来,map中没有,要加的时候,im也过来了,map中也没有,im也要加?)
     * 答案是:根本没有可能！同步方法！
     * mapState:是否共享状态机
     */
    private synchronized void doUpdateForDetailStatus(List<MessageRecallVo> vos, String code, boolean mapState) {
        log.info("开始执行消息状态更新方法:doUpdateForDetailStatus......");
        List<Long> sendedMessage = new ArrayList<>();
        try {
            vos.forEach(vo -> {
                //共享状态机,执行状态机逻辑
                if (mapState) {
                    //判断id在不在map中
                    if (detailIdMaps.containsKey(vo.getDetailId().toString())) {
                        //如果在,说明IM已经处理完了,已经在map中了;(或者rdb已经处理完了,已经在map中了)
                        //且不为false,加入到批量修改状态队列
                        if (!detailIdMaps.get(vo.getDetailId().toString()).equals("false")) {
                            sendedMessage.add(vo.getDetailId());
                            //清理map
                            detailIdMaps.remove(vo.getDetailId().toString());
                        }
                    }
                    //如果不在,加入到map中
                    else {
                        detailIdMaps.put(vo.getDetailId().toString(), "true");
                    }
                }
                //不共享状态,直接add
                else {
                    sendedMessage.add(vo.getDetailId());
                }
            });
            //批量更新,这里code写死为sended(因为map中有的都是IM成功处理的)
            if (!sendedMessage.isEmpty()) {
                //如果code是sended=2,说明要修改sended状态
                if (code.equals(MessageStatus.SENDED.getCode())) {
                    if (biliUserChatDetailMapper.updateByIdBatch(sendedMessage, code) < 1) {
                        log.error("更新失败呀？");
                        throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
                    }
                }
                //如果code是1,说明修改是离线状态
                else if (code.equals(IsFlagEnum.YES.getCode()) || code.equals(IsFlagEnum.NO.getCode())) {
                    if (biliUserChatDetailMapper.updateOffByIdBatch(sendedMessage, code, MessageStatus.SENDED.getCode()) < 1) {
                        log.error("更新离线失败呀?");
                        throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
                    }
                }
            }
        } catch (
                Exception e) {
            log.error("处理id时,未知异常:{}", e.toString());
            //报错了,恢复map
            if (!sendedMessage.isEmpty()) {
                sendedMessage.stream().parallel().forEach(id -> detailIdMaps.put(id.toString(), "true"));
            }
            throw new CustomException(AppHttpCodeEnum.SERVER_ERROR);
        }

    }

    //定时任务,一波清理map中的false
    @XxlJob("map-handler")
    public void clearMap() {
        //清理掉false键盘
        detailIdMaps.forEach((k, v) -> {
            if (v.equals("false")) {
                if (!detailIdMaps.remove(k, v)) {
                    log.error("怎么删不掉呀[{},{}]", k, v);
                }
            }
        });
    }
}
